package com.wg.core.thread.fiber;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.wg.core.lang.IDisposable;
import com.wg.core.thread.CoreThreadPool;
import com.wg.core.thread.IThreadPool;
import com.wg.core.thread.scheduler.Scheduler;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 基于线程池的轻量级任务调度组件
 */
public class PoolFiber implements IFiber {
    private static final Log log = LogFactory.get();
    /**
     * 默认批处理大小
     */
    private static final int DEFAULT_MAX_BATCH_SIZE = 100;
    /**
     * 默认队列容量（10万）
     */
    private static final int DEFAULT_QUEUE_CAPACITY = 100_000;
    /**
     * 状态
     */
    private final AtomicReference<State> state = new AtomicReference<>(State.STOPPED);
    /**
     * 线程池
     */
    private final IThreadPool pool;
    /**
     * 调度器
     */
    private final Scheduler scheduler;
    /**
     * 任务队列（有界队列）
     */
    private final BlockingQueue<Runnable> taskQueue;
    /**
     * 正在消费
     */
    private final AtomicBoolean isFlushing = new AtomicBoolean(false);
    /**
     * 批处理大小限制
     */
    private final int maxBatchSize;
    /**
     * 队列容量
     */
    private final int queueCapacity;
    /**
     * 拒绝策略
     */
    private final RejectedTaskHandler rejectedHandler;
    /**
     * 本地任务队列
     */
    private static final ThreadLocal<List<Runnable>> localActions = ThreadLocal
            .withInitial(() -> new ArrayList<>(DEFAULT_MAX_BATCH_SIZE));
    /**
     * 销毁时是否清空任务队列
     */
    private final boolean clearTaskQueueOnDispose;

    enum State {
        STARTING, RUNNING, STOPPED
    }

    /**
     * 默认构造：使用核心线程池、默认批处理大小、默认队列容量、阻塞拒绝策略
     */
    public PoolFiber() {
        this(CoreThreadPool.getInstance(), DEFAULT_MAX_BATCH_SIZE, false,
                DEFAULT_QUEUE_CAPACITY, RejectedTaskHandler.BLOCKING);
    }

    /**
     * 自定义线程池
     */
    public PoolFiber(IThreadPool pool) {
        this(pool, DEFAULT_MAX_BATCH_SIZE, false,
                DEFAULT_QUEUE_CAPACITY, RejectedTaskHandler.BLOCKING);
    }

    /**
     * 全参数构造
     * @param pool 线程池
     * @param maxBatchSize 批处理大小
     * @param clearTaskQueueOnDispose 销毁时是否清空队列
     * @param queueCapacity 队列容量
     * @param rejectedHandler 拒绝策略
     */
    public PoolFiber(IThreadPool pool, int maxBatchSize, boolean clearTaskQueueOnDispose,
                    int queueCapacity, RejectedTaskHandler rejectedHandler) {
        this.pool = pool;
        this.scheduler = new Scheduler(this);
        this.maxBatchSize = Math.max(maxBatchSize, DEFAULT_MAX_BATCH_SIZE); 
        this.clearTaskQueueOnDispose = clearTaskQueueOnDispose;
        this.queueCapacity = Math.max(queueCapacity, 1000); 
        this.taskQueue = new ArrayBlockingQueue<>(this.queueCapacity); 
        this.rejectedHandler = rejectedHandler != null ? rejectedHandler : RejectedTaskHandler.BLOCKING;
    }

     /**
     * 【仅用于拒绝策略】尝试将任务加入队列（非阻塞）
     * 注意：此方法仅应在任务被首次拒绝后，由RejectedTaskHandler调用
     * @param task 任务
     * @return 成功返回true，队列满返回false
     */
    public boolean tryOfferTask(Runnable task) {
        return taskQueue.offer(task);
    }

    /**
     * 【仅用于拒绝策略】尝试将任务加入队列（带超时）
     * 注意：此方法仅应在任务被首次拒绝后，由RejectedTaskHandler调用
     * @param task 任务
     * @param timeout 超时时间
     * @param unit 时间单位
     * @return 成功返回true，超时返回false
     * @throws InterruptedException 线程被中断
     */
    public boolean tryOfferTaskWithTimeout(Runnable task, long timeout, TimeUnit unit) throws InterruptedException {
        return taskQueue.offer(task, timeout, unit);
    }

    @Override
    public void dispose() {
        stop();
    }

    @Override
    public void start() {
        if (!state.compareAndSet(State.STOPPED, State.STARTING)) {
            throw new IllegalThreadStateException("Fiber already started or stopping");
        }
        state.set(State.RUNNING);
        enqueue(() -> {}); // 触发首次flush
    }

    @Override
    public void stop() {
        if (state.getAndSet(State.STOPPED) == State.STOPPED) {
            return;
        }
        scheduler.dispose(); 
        if (clearTaskQueueOnDispose) {
            taskQueue.clear(); 
        }
    }

    @Override
    public boolean isRunning() {
        return state.get() == State.RUNNING;
    }

    @Override
    public void enqueue(Runnable task) {
        if (state.get() == State.STOPPED) {
            return;
        }

        // 尝试入队，失败则触发拒绝策略
        boolean added = taskQueue.offer(task);
        if (!added) {
            rejectedHandler.rejected(task, this);
            return;
        }

        // 检查是否需要触发flush
        if (state.get() == State.RUNNING && isFlushing.compareAndSet(false, true)) {
            pool.execute(this::flush);
        }

        // 队列使用率超过80%时预警
        if (getQueueUsage() > 0.8) {
            log.warn("Queue is almost full. Usage: {}/{} ({:.1f}%)",
                    taskQueue.size(), queueCapacity, getQueueUsage() * 100);
        }
    }

    /**
     * 批量执行任务
     */
    private void flush() {
        List<Runnable> actions = localActions.get();
        actions.clear();

        taskQueue.drainTo(actions, maxBatchSize);

        if (actions.isEmpty()) {
            isFlushing.set(false);
            return;
        }

        try {
            for (Runnable task : actions) {
                try {
                    task.run();
                } catch (Exception e) {
                    log.error("Task execution failed", e);
                }
            }
        } finally {
            // 处理剩余任务
            if (!taskQueue.isEmpty() && state.get() == State.RUNNING) {
                pool.execute(this::flush); // 继续提交flush任务
            } else {
                isFlushing.set(false);
                // 双重检查，避免漏提交
                if (!taskQueue.isEmpty() && isFlushing.compareAndSet(false, true)) {
                    pool.execute(this::flush);
                }
            }
        }
    }

    /**
     * 获取剩余任务数
     */
    public long getRemainCount() {
        return taskQueue.size();
    }

    /**
     * 获取队列容量
     */
    public int getQueueCapacity() {
        return queueCapacity;
    }

    /**
     * 获取队列使用率（0.0 ~ 1.0）
     */
    public double getQueueUsage() {
        return (double) taskQueue.size() / queueCapacity;
    }

    /**
     * 检查队列是否接近满（使用率>80%）
     */
    public boolean isQueueAlmostFull() {
        return getQueueUsage() > 0.8;
    }

    @Override
    public IThreadPool getThreadPool() {
        return pool;
    }

    @Override
    public IDisposable schedule(Runnable task, int firstInMs) {
        if (state.get() != State.RUNNING) {
            throw new IllegalStateException("Fiber is not running");
        }
        return scheduler.schedule(task, firstInMs);
    }

    @Override
    public IDisposable scheduleOnInterval(Runnable task, int firstInMs, int regularInMs) {
        if (state.get() != State.RUNNING) {
            throw new IllegalStateException("Fiber is not running");
        }
        return scheduler.scheduleOnInterval(task, firstInMs, regularInMs);
    }
}